package com.bw.app.ods;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.bw.app.function.CustomerDeserialization;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.KafkaProducer;

/**
 * @author whd
 */
public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        //设置访问 HDFS 的用户名
        System.setProperty("HADOOP_USER_NAME", "root");

        //TODO 1.env 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.1 设置CK&状态后端 ,如下代码自己做测试不需要不然每次都要打开HDFS，未来在生产环境中写代码时这些东西一定是要有的
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop101:8020/gmall-flink-2021/ck"));
        //开启 Checkpoint,每隔 5 秒钟做一次 CK  头和头
        //env.enableCheckpointing(5000L);
        //指定 CK 的一致性语义
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);//尾和头

        //TODO 2.source 通过FlinkCDC构建SourceFunction并读取数据
        //builder是泛型方法  <String> 这个泛型代表你读取数据要封装成什么类型
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("hadoop100")//主机名
                .port(3306)//mysql端口
                .username("root")//mysql用户名
                .password("root")//mysql密码
                .databaseList("gmall-2021-flink")//监控的数据库，可以监控多个库
                //.tableList("gmall-2021-flink.z_user_info")//可以指定多个，可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
                .deserializer(new CustomerDeserialization())//反序列化方式
                //.startupOptions(StartupOptions.initial())//在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的binlog
                .startupOptions(StartupOptions.latest())//在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的binlog
                //之前我们先建库再建表然后开启binlog
                //如果要想用earliest()，必须要先修改my.cnf 先把库名写好，然后再去创建库、表，它必须要读到你的建表语句，必须要包含完整的binlog才可以
                //latest() 是直接读取最新的binlog，而不会去打印老的
                //specificOffset() 指定offset读取
                //timestamp() 指定时间戳，每条数据有自己的时间      大于等于指定时间戳开始读
                //总结:用的比较多的是initial() 和 latest()
                .build();//构建SourceFunction

        //使用 CDC Source 从 MySQL 读取数据
        DataStreamSource<String> streamSource = env.addSource(mysqlSource);

        //3.打印数据并将数据写入Kafka
        streamSource.print();
        //读取mysql数据写到ods，再用flink消费数据写到dwd，然后又消费dwd写到dwm
        //每次new 不太方便，所以可以将代码提取出来放入到工具类中
        //streamSource.addSink(new FlinkKafkaProducer<String>("hadoop101:9092","ods_base_db",new SimpleStringSchema()));
        String sinkTopic = "ods_base_db";
        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //4.启动任务
        env.execute("FlinkCDC");

    }
}
